package com.itheima.kafka.boot.stream;

import com.itheima.kafka.boot.config.stream.KafkaStreamListener;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
 * @Description:
 * @Version: V1.0
 */
@Component
public class StreamListener implements KafkaStreamListener<KStream<String,String>> {

    /**
     * from 接收待处理的消息（无界）
     * @return
     */
    @Override
    public String listenerTopic() {
        return "itcast-topic-input";
    }

    /**
     * to 发送处理之后的数据
     * @return
     */
    @Override
    public String sendTopic() {
        return "itcast-topic-output";
    }

    /**
     * 流式数据处理 processor
     * @param stream
     * @return
     */
    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        return  stream.flatMapValues((key,value)->{
            System.out.println("接收待处理数据： " + value);
            return Arrays.asList(value.split(" "));
        })
                .groupBy((key,value) -> {
                    System.out.println("前：key: "+key+"-- value:"+value);
                    return value;
                })
                .windowedBy(TimeWindows.of(5000))
                .count()
                .toStream()
                .map((key,value)->{
                    return new KeyValue<>(key.key().toString(), value.toString());
                });
    }
}
